/**
 * 
 */
package edu.ucr.cs.dblab.acestore.base.heapfile;

import java.io.IOException;
import java.nio.ByteBuffer;

import edu.ucr.cs.dblab.acestore.base.ACEContext;
import edu.ucr.cs.dblab.acestore.base.LRUCacheAlgorithm;
import edu.ucr.cs.dblab.acestore.base.datatypes.ISerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.IntegerSerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.SchemaDescriptor;
import edu.ucr.cs.dblab.acestore.base.datatypes.TupleBuilder;

/**
 * 
 * A simple implementation of the heap file index structure.
 * 
 * The structure of this index is as follows:
 * 
 * +----------+-----------+-----------+----
 * | HeadPage | HeapPage0 | HeapPage1 | ... 
 * +----------+-----------+-----------+----
 * 
 * The head page contains the following information:
 * 
 * +---------------------+---------------+
 * | AvailablePageID:int | PageCount:int |
 * +---------------------+---------------+
 * 
 * And a heap page has the following format:
 * 
 * +------+-----+------+-------+-------------+-----+-------------+--------------+--------------+    +-----+<br>
 * | Rec0 | ... | Recn | blank | offsetn:int | ... | offset0:int | recCount:int | nextPage:int-|--> | ... |<br> 
 * +------+-----+------+-------+-------------+-----+-------------+--------------+--------------+    +-----+<br>
 * 
 * And each record is formated as follows:
 * 
 * +------------------+--------------------+--------------------+-----+--------------------+--------+--------+-----+--------+ 
 * | FieldCount:short | Field0EndOfs:short | Field1EndOfs:short | ... | FieldnEndOfs:short | Field0 | Field1 | ... | Fieldn |
 * +------------------+--------------------+--------------------+-----+--------------------+--------+--------+-----+--------+
 * 
 * Currently it is an append-only index. TODO The deletion/update
 * features may be added in the future.
 * 
 * @author jarod
 * @created Jan 18, 2011
 */
public class SimpleHeapFile {

	private final ACEContext ctx;
	
	protected final SchemaDescriptor schemaDescriptor;
	
	private int idxID;
	
	private int pageCount;
	
	private int availablePageID;
	
	private static int INTSIZE = Integer.SIZE / 8;
	
	private static int SHORTSIZE = Short.SIZE / 8;
	
	private boolean updateHead;
	
	public SimpleHeapFile(ACEContext ctx, SchemaDescriptor schemaDescriptor){
		this.ctx = ctx;
		this.schemaDescriptor = schemaDescriptor;
		this.idxID = -1;
		availablePageID = -1;
		pageCount = 0;
		updateHead = false;
	}
	
	public SimpleHeapFile(ACEContext ctx, SchemaDescriptor schemaDescriptor, int idxID) throws IOException{
		this.ctx = ctx;
		this.schemaDescriptor = schemaDescriptor;
		this.idxID = idxID;
		initializeFromHeadPage();
		updateHead = false;
	}
	
	private void initializeFromHeadPage() throws IOException{
		ByteBuffer page = ctx.readPage(idxID, 0);
		availablePageID = page.getInt(0);
		pageCount = page.getInt(INTSIZE);
	}
	
	/**
	 * Parse a given record. Here the record is supposed to have the following format:
	 * 
	 * 
	 * @param rec
	 * @return
	 * @throws IOException
	 */
	public String parseRecord(ByteBuffer rec) throws IOException{
		StringBuilder sbder = new StringBuilder();
		int offset = 0;
		int fieldCount = rec.getShort(offset);
		offset += (1 + fieldCount) * SHORTSIZE;
		for(int i = 0; i < fieldCount; i++){
			sbder.append(schemaDescriptor.getFields()[i].deserialize(
					rec.array(), 
					offset + (i == 0 ? 0 : rec.getShort(SHORTSIZE * i)), 
					rec.getShort(SHORTSIZE * (i + 1)) - (i == 0 ? 0 : rec.getShort(SHORTSIZE * i))));
			sbder.append("\t");
		}
		return sbder.toString();
	}
	
	/**
	 * Insert a new piece of data.
	 * 
	 * @param data
	 * @throws IOException
	 */
	public void insert(byte[] data) throws IOException{
		int pid = availablePageID;
		ByteBuffer page;
		if(pid < 0){
			// Allocate new file for the index
			this.idxID = ctx.allocateIndex();
			// Allocate head page
			allocatePageInMem();
			// Invoked when the index is created and no page has been assigned
			page = allocatePageInMem();
			// Update the current visiting page id
			pid = pageCount - 1;
			// Assign the available page ID
			availablePageID = pid;
		}else{
			page = ctx.readPage(idxID, pid);
			while(freeSpace(page) < data.length){
				pid = getNextPage(page);
				if(pid < 0){
					// Allocate new page
					ByteBuffer newpage = allocatePageInMem();
					// Update the pointer to the new page
					page.putInt(page.capacity() - INTSIZE, pageCount - 1);
					page = newpage;
					// Update the current visiting page id
					pid = pageCount - 1;
					// Write the page content back to the memory
					ctx.writePage(idxID, pid, page);
				} else
					page = ctx.readPage(idxID, pid);
			}
			this.availablePageID = pid;
		}
		// Add data
		int recCount = getRecCount(page);
		int offset = recCount == 0 ? 0 : getEndOffset(page, recCount - 1) + 1;
		page.position(offset);
		page.put(data, 0, data.length);
		page.position(0);
		// Add end offset
		page.putInt(page.capacity() - INTSIZE * 2 - INTSIZE * (recCount + 1), offset + data.length - 1);
		// Update record count
		page.putInt(page.capacity() - INTSIZE * 2, recCount + 1);
		// Write the page back to the memory
		ctx.writePage(idxID, pid, page);
		// Write the head page back
		writeHeadBack();
	}
	
	/**
	 * Query a record based on the page id and the record id.
	 * 
	 * @param pageID
	 * @param recID
	 * @return
	 * @throws IOException
	 */
	public ByteBuffer query(int pageID, int recID) throws IOException{
		ByteBuffer page = ctx.readPage(idxID, pageID);
		int offset = (recID == 0 ? 0 : getEndOffset(page, recID - 1) + 1);
		int length = getEndOffset(page, recID) - offset + 1;
		byte[] data = new byte[length];
		page.position(offset);
		page.get(data, 0, length);
		page.position(0);
		return ByteBuffer.wrap(data);
	} 
	
	/**
	 * Write the meta information back to the head page.
	 * 
	 * @throws IOException
	 */
	private void writeHeadBack() throws IOException{
		if (updateHead) {
			ByteBuffer head = ctx.readPage(idxID, 0);
			head.position(0);
			head.putInt(availablePageID);
			head.putInt(INTSIZE, pageCount);
			ctx.writePage(idxID, 0, head);
		}
	}
	
	private int getEndOffset(ByteBuffer page, int recid){
		return page.getInt(page.capacity() - INTSIZE * 2 - INTSIZE * (recid + 1));
	}
	
	private int getRecCount(ByteBuffer page){
		return page.getInt(page.capacity() - INTSIZE * 2);
	}
	
	private ByteBuffer allocatePageInMem() throws IOException{
		this.pageCount ++;
		updateHead = true;
		ByteBuffer newpage = ctx.allocatePage(idxID, pageCount - 1);
		newpage.putInt(newpage.capacity() - INTSIZE, -1);
		return newpage;
	}
	
	private int getNextPage(ByteBuffer page) throws IOException{
		return page.getInt(page.capacity() - INTSIZE );
	}
	
	/**
	 * Check the free space of a give page.
	 * 
	 * @param page
	 * @return
	 * @throws IOException
	 */
	private int freeSpace(ByteBuffer page) throws IOException{
		return page.capacity() 
		- page.getInt(page.capacity() - INTSIZE * 2) * (schemaDescriptor.getSize() + INTSIZE + (schemaDescriptor.getFields().length + 1) * SHORTSIZE)
		- INTSIZE * 2;
	}
	
	public void close() throws IOException{
		this.writeHeadBack();
	}
	
	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		ACEContext ctx = new ACEContext(512, 3, new LRUCacheAlgorithm());
		SimpleHeapFile test1 = new SimpleHeapFile(ctx, new SchemaDescriptor(new ISerDeserializer[] {IntegerSerDeserializer.getSingleton()}));
		for(int i = 0; i < 2000; i++)
			test1.insert(TupleBuilder.BUILDTUPLE(new byte[][]{IntegerSerDeserializer.SERIALIZE(i)}));
		System.out.println(test1.parseRecord(test1.query(1,0)));
		System.out.println(test1.parseRecord(test1.query(2,0)));
		System.out.println(ctx.getIOStat());
		System.out.println("Meta Data:\n" + ctx.getIndexList());
		String dbLoc = ctx.getLocation();
		System.out.println(test1.parseRecord(test1.query(1,3)));
		System.out.println("Meta Data:\n" + ctx.getIndexList());
		test1.close();
		ctx.close();
		
		ctx = new ACEContext(3, new LRUCacheAlgorithm(), dbLoc);
		test1 = new SimpleHeapFile(ctx, new SchemaDescriptor(new ISerDeserializer[] {IntegerSerDeserializer.getSingleton()}), 0);
		for(int i = 0; i < 1; i++){
			System.out.println(test1.parseRecord(test1.query(1,i)));
			System.out.println(test1.parseRecord(test1.query(1,i)));
			System.out.println(test1.parseRecord(test1.query(2,i)));
			System.out.println(test1.parseRecord(test1.query(3,i)));
			System.out.println(test1.parseRecord(test1.query(1,i)));
			System.out.println(test1.parseRecord(test1.query(4,i)));
			System.out.println(test1.parseRecord(test1.query(1,i)));
		}
		System.out.println(ctx.getIOStat());
		System.out.println("Meta Data:\n" + ctx.getIndexList());
		test1.close();
		ctx.close();
	}

}
